package cn.itcast.streaming.function.flatmap;

import cn.itcast.streaming.bean.CustomRuleAlarmResultModel;
import cn.itcast.streaming.bean.ItcastDataPartObj;
import cn.itcast.streaming.bean.RuleInfoModel;
import cn.itcast.streaming.bean.VehicleInfoModel;
import cn.itcast.streaming.utils.CustomRuleUtil;
import com.google.common.collect.Lists;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.co.CoFlatMapFunction;
import org.apache.flink.util.Collector;

import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Objects;

//自定义告警规则广播流与拉宽后的原始数据流进行业务逻辑处理
public class CustomRuleAlarmExpressionFunction implements CoFlatMapFunction<
        ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>>
        ,ArrayList<RuleInfoModel>,
        ArrayList<CustomRuleAlarmResultModel>> {
    private ArrayList<RuleInfoModel> ruleInfoModels=new ArrayList<>();

    @Override
    public void flatMap1(ArrayList<Tuple2<ItcastDataPartObj, VehicleInfoModel>> value, Collector<ArrayList<CustomRuleAlarmResultModel>> collector) throws Exception {
        ArrayList<CustomRuleAlarmResultModel> resultModels = new ArrayList<>();
        //每条规则数据和原始数据（多对多关系）
        try {
            ruleInfoModels.forEach(ruleInfoModel -> {
                value.forEach(tuple -> {
                    ItcastDataPartObj itcastDataPartObj = tuple.f0;
                    VehicleInfoModel vehicleInfoModel = tuple.f1;
                    //定义返回的自定义告警规则结果的数据模型对象
                    CustomRuleAlarmResultModel customRuleAlarmResultModel = null;
                    //每个自定义告警规则表达式包含2-4个字段
                    //判断是否存在第三个字段，如果存在说明存在两个计算表达式，如果只有两个字段说明存在一个计算表达式
                    if (validateExistsField(ruleInfoModel.getAlarmParam3Field(), itcastDataPartObj)) {
                        //存在第三个字段，表示处理两个表达式，1和2一个表达式，第三个和可能存在的第四个字段构成一个表达式
                        //如果第一个和第二个字段在bean中都存在，且属性值是double类型，说明两个字段都是正真确的
                        boolean expressionResult1 = false;
                        boolean expressionResult2 = false;
                        try {
                            if (validateExistsField(ruleInfoModel.getAlarmParam1Field(), itcastDataPartObj) && validateExistsField(ruleInfoModel.getAlarmParam2Field(), itcastDataPartObj)) {
                                expressionResult1 = CustomRuleUtil.expression(
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam1Field())),
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam2Field())),
                                        ruleInfoModel.getOperator1(),
                                        ruleInfoModel.getAlarmThreshold1(),
                                        ruleInfoModel.getRuleSymbol1(),
                                        ruleInfoModel.getAlarmThreshold2(),
                                        ruleInfoModel.getRuleSymbol2(),
                                        ruleInfoModel.getLogicalSymbol()
                                );
                            }
                            //判断第四个字段是否存在
                            if (validateExistsField(ruleInfoModel.getAlarmParam4Field(), itcastDataPartObj)) {
                                //第二个表达式的计算参数是完整的
                                    expressionResult2 = CustomRuleUtil.expression(
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam3Field())),
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam4Field())),
                                        ruleInfoModel.getOperator2(),
                                        ruleInfoModel.getAlarmThreshold3(),
                                        ruleInfoModel.getRuleSymbol3(),
                                        ruleInfoModel.getAlarmThreshold4(),
                                        ruleInfoModel.getRuleSymbol4(),
                                        ruleInfoModel.getLogicalSymbol2()
                                );
                            } else {
                                //第二个表达式只有第三个参数
                                expressionResult2 = CustomRuleUtil.expression(
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam3Field())),
                                        0,
                                        ruleInfoModel.getOperator2(),
                                        ruleInfoModel.getAlarmThreshold3(),
                                        ruleInfoModel.getRuleSymbol3(),
                                        ruleInfoModel.getAlarmThreshold4(),
                                        ruleInfoModel.getRuleSymbol4(),
                                        null
                                );
                            }
                        } catch (IllegalAccessException e) {
                            e.printStackTrace();
                        } catch (InvocationTargetException e) {
                            e.printStackTrace();
                        } catch (NoSuchMethodException e) {
                            e.printStackTrace();
                        }

                        System.out.println("规则名称:"+ruleInfoModel.getRuleName()+"两个表达式的计算结果分别为"+expressionResult1+"和"+expressionResult2);
                        //todo 5.根据告警类型的维度进行判断触发（根据计算结果（告警或非告警）
                        if (ruleInfoModel.getMonitorTypeId().equals("1")) {
                            //todo 5.1 全部车辆
                            customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult1 && expressionResult2);
                        } else if (ruleInfoModel.getMonitorTypeId().equals("2") && vehicleInfoModel.getModelCode().equals(ruleInfoModel.getModelCode())) {
                            //todo 5.2车型告警
                            customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult1 && expressionResult2);
                        } else if (ruleInfoModel.getMonitorTypeId().equals("3") && itcastDataPartObj.getProvince().equals(ruleInfoModel.getProvince()) && itcastDataPartObj.getCity().equals(ruleInfoModel.getCity())) {
                            //todo 5.3区域告警
                            customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult1 && expressionResult2);
                        } else if (ruleInfoModel.getMonitorTypeId().equals("4") || ruleInfoModel.getMonitorTypeId().equals("5")) {
                            //todo 5.4大客户告警,车型告警
                            ArrayList<String> vins = Lists.newArrayList(ruleInfoModel.getVins().split(","));
                            if (vins.contains(itcastDataPartObj.getVin())) {
                                customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult1 && expressionResult2);
                            }
                        } else if (ruleInfoModel.getMonitorTypeId().equals("6") && vehicleInfoModel.getSeriesCode().equals(ruleInfoModel.getSeriesCode())) {
                            //todo 5.5车系告警
                            customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult1 && expressionResult2);
                        }
                    } else {
                        //没有存在第三个字段，因此只有一个表达式,如果第一个和第二个字段在bean中都存在，且属性值是double类型，说明两个字段都是正真确的
                        if (validateExistsField(ruleInfoModel.getAlarmParam1Field(), itcastDataPartObj) && validateExistsField(ruleInfoModel.getAlarmParam2Field(), itcastDataPartObj)) {
                            //只有一个规则表达式的情况
                            try {
                                boolean expressionResult = CustomRuleUtil.expression(
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam1Field())),
                                        Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj, ruleInfoModel.getAlarmParam2Field())),
                                        ruleInfoModel.getOperator1(),
                                        ruleInfoModel.getAlarmThreshold1(),
                                        ruleInfoModel.getRuleSymbol1(),
                                        ruleInfoModel.getAlarmThreshold2(),
                                        ruleInfoModel.getRuleSymbol2(),
                                        ruleInfoModel.getLogicalSymbol()
                                );
                                System.out.println("规则名称:"+ruleInfoModel.getRuleName()+" 只有一个表达式的计算结果：" + expressionResult);
                                //todo 5.根据告警类型的维度进行判断触发（根据计算结果（告警或非告警）
                                if (ruleInfoModel.getMonitorTypeId().equals("1")) {
                                    //todo 5.1 全部车辆
                                    customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult);
                                } else if (ruleInfoModel.getMonitorTypeId().equals("2") && vehicleInfoModel.getModelCode().equals(ruleInfoModel.getModelCode())) {
                                    //todo 5.2车型告警
                                    customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult);
                                } else if (ruleInfoModel.getMonitorTypeId().equals("3") && itcastDataPartObj
                                        .getProvince().equals(ruleInfoModel.getProvince()) && itcastDataPartObj.getCity().equals(ruleInfoModel.getCity())) {
                                    //todo 5.3区域告警
                                    customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult);
                                } else if (ruleInfoModel.getMonitorTypeId().equals("4") || ruleInfoModel.getMonitorTypeId().equals("5")) {
                                    //todo 5.4大客户告警,车型告警
                                    ArrayList<String> vins = Lists.newArrayList(ruleInfoModel.getVins().split(","));
                                    if (vins.contains(itcastDataPartObj.getVin())) {
                                        customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult);
                                    }

                                } else if (ruleInfoModel.getMonitorTypeId().equals("6") && vehicleInfoModel
                                        .getSeriesCode().equals(ruleInfoModel.getSeriesCode())) {
                                    //todo 5.5车系告警
                                    customRuleAlarmResultModel = convertCustomRuleAlarmResultModel(itcastDataPartObj, vehicleInfoModel, ruleInfoModel, expressionResult);
                                }
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            } catch (InvocationTargetException e) {
                                e.printStackTrace();
                            } catch (NoSuchMethodException e) {
                                e.printStackTrace();
                            }
                        }
                    }

                    //判断返回值对象是否为空
                    if (Objects.nonNull(customRuleAlarmResultModel)) {
                        //不为空,将计算结果追加到集合
                        resultModels.add(customRuleAlarmResultModel);
                    }
                });
            });
        }catch (Exception e){
            e.printStackTrace();
        }

        //返回结果
        collector.collect(resultModels);
    }

    /**
     * 判断feildName是否在javabean中存在
     * 1.属性名没有填写，说明没有该字段
     * 2.属性名填写了，但是字段名是错误的，说明没有该字段
     * 3.属性名填写正确了，但是属性名对应的值是字符串类型的，说明不存在
     * @param fieldName
     * @param itcastDataPartObj
     * @return
     */
    private boolean validateExistsField(String fieldName,ItcastDataPartObj itcastDataPartObj){
        try {
            //判断某个字段是否在某个bean中存在
            Double.parseDouble(BeanUtils.getProperty(itcastDataPartObj,fieldName));
            return true;
        } catch (Exception e) {
            return false;
        }
    }

    /**
     * 传递拉宽所需要的维度数据，返回拉宽后的结果数据
     * @param itcastDataPartObj  原始数据（拉宽前）
     * @param vehicleInfoModel   车辆维度数据
     * @param ruleInfoModel      告警规则数据
     * @param alarmFlag
     * @return
     */
    public CustomRuleAlarmResultModel convertCustomRuleAlarmResultModel(ItcastDataPartObj itcastDataPartObj, VehicleInfoModel vehicleInfoModel, RuleInfoModel ruleInfoModel, boolean alarmFlag) {
        //1：触发报警  0：未触发
        /*int alarmResult = 0;
        if(alarmFlag){
            alarmResult = 1;
        }*/
        return new CustomRuleAlarmResultModel(
                itcastDataPartObj.getVin(),
                itcastDataPartObj.getTerminalTime(),
                alarmFlag?1:0,                         //0：未触发 1：触发报警
                Integer.parseInt(ruleInfoModel.getAlarmFrame()), //触发写入数据库的阈值
                ruleInfoModel.getRuleName(),
                vehicleInfoModel.getSeriesName(),
                vehicleInfoModel.getSeriesCode(),
                vehicleInfoModel.getModelName(),
                vehicleInfoModel.getModelCode(),
                itcastDataPartObj.getProvince(),
                itcastDataPartObj.getCity(),
                itcastDataPartObj.getCountry(),
                itcastDataPartObj.getDistrict(),
                itcastDataPartObj.getAddress(),
                itcastDataPartObj.getLat(),
                itcastDataPartObj.getLng(),
                Integer.parseInt(ruleInfoModel.getMonitorTypeId())
        );
    }

    @Override
    public void flatMap2(ArrayList<RuleInfoModel> ruleInfoModels, Collector<ArrayList<CustomRuleAlarmResultModel>> collector) throws Exception {
        this.ruleInfoModels = ruleInfoModels;
    }
}
